package com.zyx.pulsardemo.functions.sdk;

import org.apache.pulsar.functions.api.Context;
import org.apache.pulsar.functions.api.Function;

import java.util.Arrays;
import java.util.List;
import java.util.Optional;

/**
 * @author Yaxi.Zhang
 * @since 2021/10/9 15:00
 * desc: 接受用户提供的关键字, 并过滤掉任何不包含该关键字的输入字符串
 */
public class KeywordFilterFunction implements Function<String, String> {

    public static final String KEYWORD_CONFIG = "keyword";
    public static final String IGNORE_CONFIG = "ignore-case";

    @Override
    public String process(String input, Context ctx) {
        // 从上下文对象中获取关键字
        Optional<Object> keyword = ctx.getUserConfigValue(KEYWORD_CONFIG);
        // 从上下文对象获取忽略大小写设置
        Optional<Object> ignoreConfig = ctx.getUserConfigValue(IGNORE_CONFIG);

        boolean ignoreCase = ignoreConfig.isPresent() ? (boolean)ignoreConfig.get(): false;

        List<String> words = Arrays.asList(input.split("\\s"));
        // 没有关键字, 什么都不能匹配
        if (!keyword.isPresent()) {
            return null;
        } // 计算每个单词, 忽略 大小写
        else if (ignoreCase && words.stream().anyMatch(s -> s.equalsIgnoreCase((String) keyword.get()))) {
            return input;
        } // 检查一个确切的匹配
        else if (words.contains(keyword.get())) {
            return input;
        }
        return null;
    }

}
